// Copyright (c) 2019 Presto Labs Pte. Ltd.
// Author: jshin

#pragma once

#include <memory>
#include <set>
#include <stdexcept>
#include <string>
#include <map>

#include "coin2/base/string_util.h"
#include "coin2/exchange/base/executor/topic_distributor.h"
#include "coin2/exchange/base/feed/subscriber.h"
#include "coin2/exchange/base/feed/update_manager.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/exchange/di/constant.h"

class JsonSchema;

namespace coin2::exchange::base::feed {

using coin2::exchange::base::symbology::IProduct;
using coin2::exchange::feed::RawParserConfig;

namespace impl {

class IFeedProcessor {
 public:
  virtual ~IFeedProcessor() = default;
  virtual const FeedUpdateManager& upd_mgr() const = 0;
  virtual FeedUpdateManager* mutable_upd_mgr() = 0;
};

class FeedParsingProcessor : public IFeedProcessor, public executor::TopicRecordSubscriber {
 public:
  FeedParsingProcessor(
      const std::string& system_name,
      const MarketExchangeApi& mea,
      const di::ConstantInjection& ci,
      const RawParserConfig& parser_config,
      const std::string& schema_path = std::string());
  ~FeedParsingProcessor() override;

  const FeedUpdateManager& upd_mgr() const override { return upd_mgr_; }
  FeedUpdateManager* mutable_upd_mgr() override { return &upd_mgr_; }

  virtual void AddProduct(const IProduct& product, IFeedSubscriber* subscriber) {
    if (subscriber_ == nullptr) {
      subscriber_ = subscriber;
    } else {
      CHECK_EQ(subscriber_, subscriber);
    }
    upd_mgr_.AddProductForFeed(product, subscriber);
  }

  virtual void RemoveProduct(const std::vector<const IProduct*>& products) {
    upd_mgr_.RemoveProductForFeed(products);
  }

  // e.g. CheckChannelFilter("{}@trade")
  // ch = "{}@trade"
  // config: ["trade"]
  // consts: trade = "{}@trade"
  virtual bool CheckChannelFilter(const std::string& ch) {
    if (parser_config_.channels().empty()) {
      return true;
    }
    for (const auto& [name, content] : channel_strings_) {
      if (content.find(ch) != std::string::npos) {
        return true;
      } else if (name.find(ch) != std::string::npos) {
        return true;
      }
    }
    return false;
  }

  void Error(std::string_view err_msg, std::string_view raw_msg) {
    if (parser_config_.continue_on_error()) {
      LOG(ERROR) << "Processor error msg: " << err_msg << ", raw payload: <" << raw_msg << ">";
    } else {
      throw std::runtime_error(
          fmt::format("Processor error msg: {}, raw payload: {}", err_msg, raw_msg));
    }
  }

  void OnTopicRecordBase(const TopicRecordPacket& packet) override {
    upd_mgr_.SetPrePostReadStats(packet);
    OnTopicRecord(packet);
  }

  static std::map<std::string, std::string> GetRequestParameters(std::string_view url) {
    std::map<std::string, std::string> parameters;
    size_t pos = url.find('?');
    if (pos == std::string::npos) return parameters;

    std::string_view para_str = url.substr(pos + 1);
    std::vector<std::string> para_pairs = Split(para_str, "&");
    for (const auto& para_pair : para_pairs) {
      std::vector<std::string> para = Split(para_pair, "=");
      CHECK_EQ(para.size(), 2);
      parameters[para[0]] = para[1];
    }

    return parameters;
  }

 protected:
  RawParserConfig parser_config_;
  FeedUpdateManager upd_mgr_;
  std::map<std::string, std::string> channel_strings_;
  std::unique_ptr<JsonSchema> schema_;
  IFeedSubscriber* subscriber_;

 private:
};

}  // namespace impl
using impl::FeedParsingProcessor;
using impl::IFeedProcessor;
}  // namespace coin2::exchange::base::feed
